package com.san.metadata.service.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.san.base.domain.CodeDO;
import com.san.base.service.CodeService;
import com.san.common.quartz.utils.QuartzManager;
import com.san.common.utils.Constant;
import com.san.common.utils.DateUtils;
import com.san.common.utils.MD5Utils;
import com.san.common.utils.PageUtils;
import com.san.common.utils.Query;
import com.san.common.utils.StringUtils;
import com.san.earlyWarning.domain.EarlyWarningDO;
import com.san.earlyWarning.service.EarlyWarningService;
import com.san.metadata.dao.BasePersonCache;
import com.san.metadata.dao.OperationDao;
import com.san.metadata.dao.OperationMetadataManager;
import com.san.metadata.domain.BusinessDataDO;
import com.san.metadata.domain.DataSourceDO;
import com.san.metadata.domain.ManageTableDO;
import com.san.metadata.domain.OperationDO;
import com.san.metadata.domain.OperationJobDO;
import com.san.metadata.domain.TechnicalColumnsDO;
import com.san.metadata.domain.TechnicalTableDO;
import com.san.metadata.service.BusinessDataService;
import com.san.metadata.service.DataSourceService;
import com.san.metadata.service.ManageMetadataService;
import com.san.metadata.service.OperationJobService;
import com.san.metadata.service.OperationMetadataService;
import com.san.metadata.service.TechnicalMetadataService;
import com.san.metadata.task.DataExchangeThread;
import com.san.metadata.task.DataProcessThread;
import com.san.metadata.task.EarlyWarningThread;
import com.san.metadata.task.FixingBusiMetadatahread;
import com.san.metadata.task.OperationMetaDataThread;
import com.san.metadata.task.WarningsThread;
import com.san.metadata.vo.BaseLegalPersonVO;
import com.san.oms.service.ModelService;
import com.san.oms.service.ScoreService;
import com.san.warnings.domain.WarningsModelDO;
import com.san.warnings.service.WarningsService;

@Service("operationMetadataService")
public class OperationMetadataServiceImpl implements OperationMetadataService {

	private static final Logger logger = LoggerFactory.getLogger(OperationMetadataServiceImpl.class);

	@Autowired
	private OperationDao operationMapper;

	@Autowired
	QuartzManager quartzManager;

	@Autowired
	TechnicalMetadataService technicalMetadtaService;

	@Autowired
	ManageMetadataService manageMetadtaService;

	@Autowired
	OperationMetadataManager operationMetadataManager;

	@Autowired
	OperationJobService operationJobService;

	@Autowired
	BusinessDataService businessDataService;

	@Autowired
	CodeService codeService;

	@Autowired
	ModelService modelService;

	@Autowired
	ScoreService scoreService;

	@Autowired
	BasePersonCache basePersonCache;

	@Autowired
	DataSourceService dataSourceService;

	@Autowired
	EarlyWarningService earlyWarningService;

	@Autowired
	WarningsService warningsService;

	/**
	 * 批量归集表数量
	 */
	private static final int PAGE_LIMIT = 2;

	/**
	 * 企查查分页条数
	 */
	private static final int QICHACHA_PAGE_SIZE = 10;

	@Override
	public OperationDO getById(Long id) {
		return operationMapper.getById(id);
	}

	@Override
	public List<OperationDO> list(Map<String, Object> map) {
		return operationMapper.list(map);
	}

	@Override
	public int count(Map<String, Object> map) {
		return operationMapper.count(map);
	}

	@Override
	public int save(Object obj) throws Exception {
		return operationMapper.save(obj);
	}

	@Override
	public int doEdit(Object obj) throws Exception {
		return operationMapper.doEdit(obj);
	}

	@Override
	public int remove(Long id) throws Exception {
		return operationMapper.remove(id);
	}

	@Override
	public int batchRemove(Long[] ids) throws Exception {
		return operationMapper.batchRemove(ids);
	}

	@Override
	public void initSchedule() throws SchedulerException {
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("operationType", Constant.OPERATION_TYPE_CYCLE);
		// 这里获取任务信息数据
		List<OperationDO> operationList = operationMapper.list(map);
		for (OperationDO taskDO : operationList) {
			if (0 == taskDO.getStatus()) {
				quartzManager.addOperationJob(taskDO);
			}
		}
	}

	@Override
	public void changeStatus(Long taskId, String cmd) throws SchedulerException, Exception {
		OperationDO taskDO = operationMapper.getById(taskId);
		if (taskDO == null) {
			return;
		}
		if (Constant.STATUS_RUNNING_STOP.equals(cmd)) {
			quartzManager.deleteOperationJob(taskDO);
			taskDO.setStatus(Constant.STATUS_INVALID);
		} else {
			if (!Constant.STATUS_RUNNING_START.equals(cmd)) {
			} else {
				taskDO.setStatus(Constant.STATUS_NORMAL);
				quartzManager.addOperationJob(taskDO);
			}
		}

		doEdit(taskDO);
	}

	@Override
	public void runOperationJob(Long operationId) {

		Map<String, Object> map = new HashMap<String, Object>();
		map.put("operationId", operationId);
		map.put("sort", "priority");
		map.put("order", "desc");
		List<OperationJobDO> list = operationJobService.list(map);

		// 启动操作元单次任务调度线程，防止作业过多，前台等待时间过长
		OperationMetaDataThread thread = new OperationMetaDataThread(list);
		thread.start();
	}

	@Override
	public void dataCollect(Long batchNo, Long jobId) throws Exception {
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("importType", Constant.IMPORT_TYPE_EXCHANGE);
		map.put("offset", 0);
		map.put("limit", PAGE_LIMIT);
		PageUtils pageUtils = technicalMetadtaService.list(new Query(map));
		int count = pageUtils.getTotal();
		int page = count % PAGE_LIMIT == 0 ? count / PAGE_LIMIT : count / PAGE_LIMIT + 1;

		ExecutorService pool = Executors.newFixedThreadPool(Constant.THREAD_COUNT);
		ArrayList<Callable<Long>> callers = new ArrayList<Callable<Long>>();

		List<TechnicalTableDO> list = (List<TechnicalTableDO>) pageUtils.getRows();

		for (TechnicalTableDO tableDO : list) {
			if (Constant.STATUS_NORMAL == tableDO.getStatus()) {
				callers.add(new DataExchangeThread(operationMetadataManager, tableDO, batchNo, jobId));
			}
		}
		List<Future<Long>> listFuture = pool.invokeAll(callers);

		for (int i = 1; i < page; i++) {
			callers = new ArrayList<Callable<Long>>();
			map = new HashMap<String, Object>();
			map.put("importType", 1);
			map.put("offset", i * PAGE_LIMIT);
			map.put("limit", PAGE_LIMIT);
			pageUtils = technicalMetadtaService.list(new Query(map));

			list = (List<TechnicalTableDO>) pageUtils.getRows();
			for (TechnicalTableDO tableDO : list) {
				if (Constant.STATUS_NORMAL == tableDO.getStatus()) {
					callers.add(new DataExchangeThread(operationMetadataManager, tableDO, batchNo, jobId));
				}
			}
			listFuture = pool.invokeAll(callers);
		}
		pool.shutdown();
	}

//	/**
//	 * 自定义数据同步
//	 * 
//	 * @param batchNo
//	 * @param jobId
//	 * @throws Exception
//	 */
//	private void dataSync(Long batchNo, Long jobId) throws Exception {
//		operationMetadataManager.dataSync(batchNo, jobId);
//	}

	@Override
	public void dataProcess(Long batchNo, Long jobId) throws Exception {
		CodeDO legalCodeDO = codeService.getByName(Constant.KEY_LEGAL_PERSON_TABLES);
		// 法人辨识来源表
		List<String> baseLegalTableList = new ArrayList<String>();
		if (null != legalCodeDO && !StringUtils.isEmpty(legalCodeDO.getCodeValue())) {
			String[] tableNames = legalCodeDO.getCodeValue().split(",");
			for (String tableName : tableNames) {
				baseLegalTableList.add(tableName.toLowerCase());
			}
		}
		CodeDO naturalCodeDO = codeService.getByName(Constant.KEY_NATURAL_PERSON_TABLES);
		// 自然人辨识来源表
		List<String> baseNaturalTableList = new ArrayList<String>();
		if (null != naturalCodeDO && !StringUtils.isEmpty(naturalCodeDO.getCodeValue())) {
			String[] tableNames = naturalCodeDO.getCodeValue().split(",");
			for (String tableName : tableNames) {
				baseNaturalTableList.add(tableName.toLowerCase());
			}
		}

		// 先归集辨识表
		this.baseTableProcess(batchNo, jobId, baseLegalTableList, baseNaturalTableList);

		Map<String, Object> map = new HashMap<String, Object>();
		map.put("importType", Constant.IMPORT_TYPE_EXCHANGE);
		map.put("offset", 0);
		map.put("limit", PAGE_LIMIT);
		PageUtils pageUtils = manageMetadtaService.list(new Query(map));
		int count = pageUtils.getTotal();
		int page = count % PAGE_LIMIT == 0 ? count / PAGE_LIMIT : count / PAGE_LIMIT + 1;

		ExecutorService pool = Executors.newFixedThreadPool(Constant.THREAD_COUNT);
		ArrayList<Callable<Long>> callers = new ArrayList<Callable<Long>>();

		List<ManageTableDO> list = (List<ManageTableDO>) pageUtils.getRows();

		for (ManageTableDO tableDO : list) {
			if (baseLegalTableList.contains(tableDO.getTargetTableName().toLowerCase())) {
				// 法人辨识表及来源表已处理，此处忽略
				continue;
			}
			if (baseNaturalTableList.contains(tableDO.getTargetTableName().toLowerCase())) {
				// 自然人辨识表及来源表已处理，此处忽略
				continue;
			}
			if (Constant.STATUS_NORMAL == tableDO.getStatus()) {
				callers.add(new DataProcessThread(operationMetadataManager, tableDO, batchNo, jobId));
			}
		}
		pool.invokeAll(callers);

		for (int i = 1; i < page; i++) {
			callers = new ArrayList<Callable<Long>>();
			map = new HashMap<String, Object>();
			map.put("importType", 1);
			map.put("offset", i * PAGE_LIMIT);
			map.put("limit", PAGE_LIMIT);
			pageUtils = manageMetadtaService.list(new Query(map));

			list = (List<ManageTableDO>) pageUtils.getRows();
			for (ManageTableDO tableDO : list) {
				if (baseLegalTableList.contains(tableDO.getTargetTableName().toLowerCase())) {
					// 法人辨识表及来源表已处理，此处忽略
					continue;
				}
				if (baseNaturalTableList.contains(tableDO.getTargetTableName().toLowerCase())) {
					// 自然人辨识表及来源表已处理，此处忽略
					continue;
				}
				if (Constant.STATUS_NORMAL == tableDO.getStatus()) {
					callers.add(new DataProcessThread(operationMetadataManager, tableDO, batchNo, jobId));
				}
			}
			pool.invokeAll(callers);
		}
		pool.shutdown();
	}

	/**
	 * 辨识表及来源表数据归集
	 * 
	 * @param batchNo
	 * @param jobId
	 * @param baseLegalTableList
	 * @param baseNaturalTableList
	 * @throws Exception
	 */
	private void baseTableProcess(Long batchNo, Long jobId, List<String> baseLegalTableList,
			List<String> baseNaturalTableList) throws Exception {
		ExecutorService pool = Executors.newFixedThreadPool(Constant.THREAD_COUNT);
		ArrayList<Callable<Long>> callers = new ArrayList<Callable<Long>>();

		basePersonCache.initLegalCache();

		// 获取法人辨识来源表，并加入待处理线程

		if (null != baseLegalTableList && baseLegalTableList.size() > 0) {
			Map<String, Object> map = new HashMap<String, Object>();
			map.put("importType", Constant.IMPORT_TYPE_EXCHANGE);
			for (String tableName : baseLegalTableList) {
				if (!StringUtils.isEmpty(tableName)) {
					map.put("targetTableName1", tableName.toUpperCase());
					PageUtils pageUtils = manageMetadtaService.list(new Query(map));
					List<ManageTableDO> list = (List<ManageTableDO>) pageUtils.getRows();

					for (ManageTableDO tableDO : list) {
						if (Constant.STATUS_NORMAL == tableDO.getStatus()) {
							callers.add(new DataProcessThread(operationMetadataManager, tableDO, batchNo, jobId));
						}
					}
				}

			}
		}

		basePersonCache.initNaturalCache();

		// 获取自然人辨识来源表，并加入待处理线程
		if (null != baseNaturalTableList && baseNaturalTableList.size() > 0) {
			Map<String, Object> map = new HashMap<String, Object>();
			map.put("importType", Constant.IMPORT_TYPE_EXCHANGE);
			for (String tableName : baseNaturalTableList) {
				if (!StringUtils.isEmpty(tableName)) {
					map.put("targetTableName1", tableName.toUpperCase());
					PageUtils pageUtils = manageMetadtaService.list(new Query(map));
					List<ManageTableDO> list = (List<ManageTableDO>) pageUtils.getRows();

					for (ManageTableDO tableDO : list) {
						if (Constant.STATUS_NORMAL == tableDO.getStatus()) {
							callers.add(new DataProcessThread(operationMetadataManager, tableDO, batchNo, jobId));
						}
					}
				}

			}
		}
		if (null != callers && callers.size() > 0) {
			// 辨识表及来源表归集
			pool.invokeAll(callers);
		}
		pool.shutdown();
	}

	@Override
	public void fixingBusiMetadata(Long batchNo, Long jobId) throws Exception {

		// 备份法人业务元数据结果表
		operationMetadataManager.bakResultTable(batchNo, Constant.BUSINESS_DATA_TYPE_LEGAL_PERSON);
		// 生成法人业务元数据结果表
		Map<String, Object> condition = new HashMap<String, Object>();
		condition.put("dataType", Constant.BUSINESS_DATA_TYPE_LEGAL_PERSON + "");
		List<BusinessDataDO> legalList = businessDataService.list(condition);
		FixingBusiMetadatahread legalThread = new FixingBusiMetadatahread(operationMetadataManager, legalList);
		legalThread.start();

		// 备份自然人人业务元数据结果表
		operationMetadataManager.bakResultTable(batchNo, Constant.BUSINESS_DATA_TYPE_NATURAL_PERSON);
		// 生成自然人业务元数据结果表
		condition = new HashMap<String, Object>();
		condition.put("dataType", Constant.BUSINESS_DATA_TYPE_NATURAL_PERSON + "");
		List<BusinessDataDO> naturalList = businessDataService.list(condition);
		FixingBusiMetadatahread naturalThread = new FixingBusiMetadatahread(operationMetadataManager, naturalList);
		naturalThread.start();
	}

	@Override
	public void qichachaCollect(Long batchNo, Long jobId) throws Exception {

		OperationJobDO operationJobDO = operationJobService.getById(jobId);

		String url = Constant.QICHACHA_DOMAIN + "/" + operationJobDO.getRequestUrl() + "&dtype=json&pageSize="
				+ QICHACHA_PAGE_SIZE;

		CloseableHttpClient client = HttpClientBuilder.create().build();

		Long technicalTableId = operationJobDO.getTechnicalTable();
		// 获取技术元数据表信息
		TechnicalTableDO technicalTableDO = technicalMetadtaService.getById(technicalTableId);
		// 获取数据源信息
		DataSourceDO dataSourceDO = (DataSourceDO) dataSourceService.getById(technicalTableDO.getTargetSourceId());
		// 获取技术元表字段信息
		List<TechnicalColumnsDO> columns = technicalMetadtaService.allTechnicalColumns(technicalTableId);

		// 写入SQL
		StringBuffer insertSQL = new StringBuffer("insert into ");
		insertSQL.append(technicalTableDO.getTargetTableName()).append(" (");
		StringBuffer paramsSQL = new StringBuffer(" (");
		for (int i = 0; i < columns.size(); i++) {
			TechnicalColumnsDO technicalColumnsDO = columns.get(i);
			insertSQL.append(technicalColumnsDO.getColumnName());
			paramsSQL.append("?");
			if (i < columns.size() - 1) {
				insertSQL.append(",");
				paramsSQL.append(",");
			}
		}
		insertSQL.append(") values ");
		paramsSQL.append(")");
		insertSQL.append(paramsSQL.toString());

		List<Object[]> allParams = null;
		PageUtils pageUtils = operationMetadataManager.getQichachBase(1);
		List list = pageUtils.getRows();

		CodeDO codeDO = codeService.getByName(Constant.QICHACHA_API_KEY);
		String apiKey = codeDO.getCodeValue();

		for (int i = 0; i < list.size(); i++) {
			Map<String, Object> obj = (Map<String, Object>) list.get(i);
			url = url.replace("<companyName>", (String) obj.get("qymc")).replace("<ApiKey>", apiKey);

			allParams = this.analysisQichachaResponse(client, url, apiKey, columns, null, allParams, 1);

			if (null != allParams) {
				operationMetadataManager.saveQichachaData(dataSourceDO.getSourceCode(), insertSQL.toString(),
						allParams);
			}
		}

		if (pageUtils.getTotal() > 10) {
			int pages = pageUtils.getTotal() % 10 == 0 ? pageUtils.getTotal() / 10 : pageUtils.getTotal() / 10 + 1;

			for (int k = 2; k <= pages; k++) {
				pageUtils = operationMetadataManager.getQichachBase(k);
				list = pageUtils.getRows();
				for (int i = 0; i < list.size(); i++) {
					Map<String, Object> obj = (Map<String, Object>) list.get(i);
					url = url.replace("<companyName>", (String) obj.get("qymc")).replace("<ApiKey>", apiKey);

					allParams = this.analysisQichachaResponse(client, url, apiKey, columns, null, allParams, 1);

					if (null != allParams) {
						operationMetadataManager.saveQichachaData(dataSourceDO.getSourceCode(), insertSQL.toString(),
								allParams);
					}
				}
			}
		}

	}

	/**
	 * 解析企查查接口返回数据
	 * 
	 * @param client
	 * @param url       企查查接口地址
	 * @param apiKey    企查查apiKey
	 * @param columns   技术元数据表字段集合
	 * @param keyMap    key值转换map，企查查key值大小写敏感，系统不敏感，需要转换
	 * @param allParams 数据解析成表参数
	 * @param page
	 * @return
	 * @throws Exception
	 */
	public List<Object[]> analysisQichachaResponse(CloseableHttpClient client, String url, String apiKey,
			List<TechnicalColumnsDO> columns, Map<String, String> keyMap, List<Object[]> allParams, int page)
			throws Exception {
		if (null == allParams) {
			allParams = new ArrayList<Object[]>();
		}

		if (null == keyMap) {
			keyMap = new HashMap<String, String>();
		}

		String response = this.getQichachaResponse(client, url, page, apiKey);

		Map result = (Map) JSONObject.parse(response);

		List<Object> params = null;

		if (null != result && "200".equals((String) result.get("Status"))) {
			JSONArray jsonArray = (JSONArray) result.get("Result");

			if (null != jsonArray) {
				for (int i = 0; i < jsonArray.size(); i++) {
					JSONObject obj = jsonArray.getJSONObject(i);

					params = new ArrayList<Object>();
					for (TechnicalColumnsDO technicalColumnsDO : columns) {
						if (Constant.NDATA_IMPORT_AUTO_INPUT_DATE
								.equals(technicalColumnsDO.getColumnName().toUpperCase())) {
							params.add(new Date());
						} else if (!StringUtils.isEmpty(keyMap.get(technicalColumnsDO.getColumnName()))) {
							params.add(obj.get(keyMap.get(technicalColumnsDO.getColumnName())));
						} else {
							for (String key : obj.keySet()) {
								if (technicalColumnsDO.getColumnName().toUpperCase().equals(key.toUpperCase())) {
									keyMap.put(technicalColumnsDO.getColumnName(), key);
									params.add(obj.get(key));
									break;
								}
							}
						}
					}

					allParams.add(params.toArray());
				}
				JSONObject paging = (JSONObject) result.get("Paging");
				if (null != paging & paging.getIntValue("TotalRecords") > page * QICHACHA_PAGE_SIZE) {
					return this.analysisQichachaResponse(client, url, apiKey, columns, keyMap, allParams, page + 1);
				}
			}
		}

		return allParams;
	}

	/**
	 * 调用企查查接口
	 * 
	 * @param client
	 * @param url
	 * @param pageIndex
	 * @param apiKey
	 * @return
	 * @throws IOException
	 * @throws ClientProtocolException
	 */
	public String getQichachaResponse(CloseableHttpClient client, String url, int pageIndex, String apiKey)
			throws IOException, ClientProtocolException {
		if (pageIndex <= 0) {
			pageIndex = 1;
		}
		HttpGet httpGet = new HttpGet(url + "&pageIndex=" + pageIndex);
		String timespan = DateUtils.getUnixTimespan();

		CodeDO codeDO = codeService.getByName(Constant.QICHACHA_SECRETKEY);

		String key = apiKey + timespan + codeDO.getCodeValue();

		String token = MD5Utils.MD5Encode(key);
		httpGet.setHeader("Token", token);
		httpGet.setHeader("Timespan", timespan);

		HttpResponse response = client.execute(httpGet);
		int statusCode = response.getStatusLine().getStatusCode();

		StringBuilder sb = new StringBuilder();
		if (statusCode == 200) {
			HttpEntity entity = response.getEntity();
			BufferedReader reader = new BufferedReader(new InputStreamReader(entity.getContent(), "UTF-8"));
			String buffer = null;
			while ((buffer = reader.readLine()) != null) {
				sb.append(buffer);
			}
		}

		return sb.toString();
	}

	@Override
	public void earlyWarningCollect(Long batchNo, Long jobId) throws Exception {
		// TODO 未完成

		// 获取需要固话的预警列表
		List<EarlyWarningDO> warnList = earlyWarningService.listAllFollow();
		EarlyWarningThread legalThread = new EarlyWarningThread(operationMetadataManager, warnList);
		legalThread.start();
	}

	@Override
	public void WarningsCollect(Long batchNo, Long jobId) throws Exception {
		// 获取需要固话的预警列表
		List<WarningsModelDO> warningsModel = warningsService.listAllFollow();
		WarningsThread legalThread = new WarningsThread(operationMetadataManager, warningsModel);
		legalThread.start();
	}
}
